package com.pw.study.flink.chapter8;

import com.pw.study.flink.entities.HotItem;
import com.pw.study.flink.entities.UserBehavior;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

public class ExerciseTop1 {


    public static void main(String[] args) {
        exercise();
    }

    private static void exercise() {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        //输入数据
        SingleOutputStreamOperator<UserBehavior> stream = env.readTextFile("data/file/UserBehavior.csv").map(ads -> {
            String[] line = ads.split(",");
            return new UserBehavior(Long.valueOf(line[0]), Long.valueOf(line[1]), Integer.valueOf(line[2]), line[3], Long.valueOf(line[4]) * 1000);
        }).filter(pv -> pv.getBehavior().equals("pv"));
        //
        WatermarkStrategy<UserBehavior> strategy = WatermarkStrategy.<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner((user, ts) -> user.getTimestamp());

        SingleOutputStreamOperator<HotItem> stream2 = stream.assignTimestampsAndWatermarks(strategy).keyBy(UserBehavior::getItemId)
                .window(SlidingEventTimeWindows.of(Time.hours(2), Time.hours(1)))
                .aggregate(new AggregateFunction<UserBehavior, Long, Long>() {
                    @Override
                    public Long createAccumulator() {
                        return 0l;
                    }

                    @Override
                    public Long add(UserBehavior value, Long acc) {
                        return acc + 1;
                    }

                    @Override
                    public Long getResult(Long acc) {
                        return acc;
                    }

                    @Override
                    public Long merge(Long a, Long b) {
                        return a + b;
                    }
                }, new ProcessWindowFunction<Long, HotItem, Long, TimeWindow>() {
                    @Override
                    public void process(Long key, Context ctx, Iterable<Long> elements, Collector<HotItem> out) throws Exception {
                        Long count = elements.iterator().next();
                        long wEnd = ctx.window().getEnd();
                        out.collect(new HotItem(key, wEnd, count));
                    }
                });
        stream2.keyBy(HotItem::getWindowEndTime).process(new KeyedProcessFunction<Long, HotItem, String>() {

                    private ListState<HotItem> hotState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        hotState = getRuntimeContext().getListState(new ListStateDescriptor<>("hotState", HotItem.class));
                    }

                    @Override
                    public void processElement(HotItem value, Context ctx, Collector<String> out) throws Exception {
                        if (!hotState.get().iterator().hasNext()) {
                            ctx.timerService().registerEventTimeTimer(value.getWindowEndTime() + 1000);
                        }
                        hotState.add(value);
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                        List<HotItem> list = StreamSupport.stream(hotState.get().spliterator(), true).collect(Collectors.toList());
                        list.sort((o1, o2) -> o2.getCount().compareTo(o1.getCount()));
                        String msg = "==================================\n ";
                        for (int i = 0, len = Math.min(3, list.size()); i < len; i++) {
                            msg += list.get(i) + "\n";
                        }
                        out.collect(msg);
                    }
                }).print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
